//
//  amqpcpp_streamer.cpp
//  Amqp_test
//
//  Created by HNXF on 2022/7/28.
//

#include "amqpcpp_streamer.h"
#include <amqpcpp.h>
#include <boost/asio/io_service.hpp>
#include <iostream>
#include "asio_connection_handler.h"
#include "sync_channel.h"


//
amqpcpp_streamer::amqpcpp_streamer()
    : ip_address_("")
    , port_(0)
    , vhost_("")
    , username_("")
    , password_("")
    , error_callback_(nullptr) {
    
}

amqpcpp_streamer::~amqpcpp_streamer() {
    
}

void amqpcpp_streamer::set_amqp_param( const std::string &addr,
                                      const uint16_t port,
                                      const std::string &vhost,
                                      const std::string &user,
                                      const std::string &pwd,
                                      const on_error_cb cb ) {
    ip_address_ = addr;
    port_ = port;
    vhost_ = vhost;
    username_ = user;
    password_ = pwd;
    error_callback_ = cb;
}

void amqpcpp_streamer::publish(
    const std::string& topic,
    const std::string& partition_key,
    const std::string& event_type_name,
                               const std::string& message) {
    try
    {
        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::publish begin\n";

        channel_->publish(topic, partition_key, event_type_name, message);

        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::publish success\n";
    }
    catch (const std::exception& e)
    {
        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::publish error: " << e.what() << "\n";
        throw;
    }
}



bool amqpcpp_streamer::connect()
{
    try
    {
        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::connect begin\n";

        stop();

        io_service_ = std::make_unique<boost::asio::io_service>();

        connection_handler_ = std::make_unique<asio_connection_handler>(*io_service_, ip_address_, port_);

        connection_ = std::make_unique<AMQP::Connection>(
            connection_handler_.get(),
            AMQP::Login(username_, password_),
            vhost_);

        channel_ = std::make_unique<sync_channel>(*io_service_, *connection_);

        io_service_thread_ = std::thread([this]()
        {
            run();
        });

        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::connect success\n";
        return true;
    }
    catch (const std::exception& e)
    {
        std::cout << std::this_thread::get_id() << ": AmqpCppStreamer::connect error: " << e.what() << "\n";
        return false;
    }
}

void amqpcpp_streamer::run()
{
    try
    {
        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::run begin\n";
        io_service_->run();
        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::run success\n";
    }
    catch (const std::exception& e)
    {
        std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::run error: " << e.what() << "\n";
        error_callback_(std::current_exception());
    }
    channel_->stop();
}

void amqpcpp_streamer::stop()
{
    std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::stop begin\n";
    if (io_service_thread_.joinable())
    {
        io_service_->stop();
        io_service_thread_.join();
    }
    std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::stop joined\n";

    channel_.reset();
    connection_.reset();
    // destruction order is important between the handler and the service
    // because some handler's internal objects depends on the service being in a valid state during destruction
    connection_handler_.reset();
    io_service_.reset();
    std::cout << std::this_thread::get_id() << ": amqpcpp_streamer::stop reset\n";
}
